package com.源码分析.thread;

import java.util.ArrayList;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import org.junit.Test;

import reactor.core.publisher.Flux;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

public class CustomThread {

	static final AtomicLong EVICTOR_COUNTER = new AtomicLong();

	static final ThreadFactory EVICTOR_FACTORY = r -> {
		Thread t = new Thread(r, "Test-" + EVICTOR_COUNTER.incrementAndGet());
		t.setDaemon(true);
		return t;
	};

	@Test
	public void test() {
		Flux.generate(ArrayList::new, (list, sink) -> {
			int value = 100;
			list.add(value);
			System.out.println("所发射元素产生的线程：" + Thread.currentThread().getName());
			sink.next(value);
			try {
				Thread.sleep(10);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			if(list.size() == 6) {
				sink.complete();
			}
			return list;
		})
		.publishOn(custom_Scheduler(), 2)
		.subscribe(x -> {
			System.out.println(x + "：" + Thread.currentThread().getName());
		});
		try {
			Thread.sleep(100000000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}


	private Scheduler custom_Scheduler() {
		Executor executor = new ThreadPoolExecutor(10, 10, 0l, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1000), EVICTOR_FACTORY);
		return Schedulers.fromExecutor(executor);
	}

}
